package base

import (
	"context"
	"crypto/tls"
	"fmt"
	log "shadow/collector/logger"
	"shadow/utils"
	"strconv"
	"strings"
	"time"

	"github.com/gomodule/redigo/redis"
)

var (
	poolClient *redis.Pool
	client     *RedisClient
)

// Options keeps the settings to setup redis connection.
type Options struct {
	Addr string

	// Use the specified Username to authenticate the current connection
	// with one of the connections defined in the ACL list when connecting
	// to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
	Username string
	// Optional password. Must match the password specified in the
	// requirepass server configuration option (if connecting to a Redis 5.0 instance, or lower),
	// or the User Password when connecting to a Redis 6.0 instance, or greater,
	// that is using the Redis ACL system.
	Password string

	// Database to be selected after connecting to the server.
	DB int

	// Maximum number of retries before giving up.
	// Default is 3 retries; -1 (not 0) disables retries.
	MaxRetries int
	// Minimum backoff between each retry.
	// Default is 8 milliseconds; -1 disables backoff.
	MinRetryBackoff time.Duration
	// Maximum backoff between each retry.
	// Default is 512 milliseconds; -1 disables backoff.
	MaxRetryBackoff time.Duration

	// Dial timeout for establishing new connections.
	// Default is 5 seconds.
	DialTimeout time.Duration
	// Timeout for socket reads. If reached, commands will fail
	// with a timeout instead of blocking. Use value -1 for no timeout and 0 for default.
	// Default is 3 seconds.
	ReadTimeout time.Duration
	// Timeout for socket writes. If reached, commands will fail
	// with a timeout instead of blocking.
	// Default is ReadTimeout.
	WriteTimeout time.Duration

	// Type of connection pool.
	// true for FIFO pool, false for LIFO pool.
	// Note that fifo has higher overhead compared to lifo.
	PoolFIFO bool
	// Maximum number of socket connections.
	// Default is 10 connections per every available CPU as reported by runtime.GOMAXPROCS.
	PoolSize int
	// Minimum number of idle connections which is useful when establishing
	// new connection is slow.
	MinIdleConns int
	// Connection age at which client retires (closes) the connection.
	// Default is to not close aged connections.
	MaxConnAge time.Duration
	// Amount of time client waits for connection if all connections
	// are busy before returning an error.
	// Default is ReadTimeout + 1 second.
	PoolTimeout time.Duration
	// Amount of time after which client closes idle connections.
	// Should be less than server's timeout.
	// Default is 5 minutes. -1 disables idle timeout check.
	IdleTimeout time.Duration
	// Frequency of idle checks made by idle connections reaper.
	// Default is 1 minute. -1 disables idle connections reaper,
	// but idle connections are still discarded by the client
	// if IdleTimeout is set.
	IdleCheckFrequency time.Duration

	// Enables read only queries on slave nodes.
	readOnly bool

	// TLS Config to use. When set TLS will be negotiated.
	TLSConfig *tls.Config
}

// RedisClient redis client
type RedisClient struct {
	pool *redis.Pool
}

func NewRedisClient(conf *Options) *RedisClient {
	client = &RedisClient{}
	client.pool = &redis.Pool{
		MaxIdle:     200,
		MaxActive:   50,
		IdleTimeout: 10 * time.Second,
		Wait:        true,
	}

	client.pool.Dial = func() (redis.Conn, error) {
		conn, err := redis.Dial("tcp", conf.Addr)
		if err != nil {
			return nil, err
		}

		password := conf.Password
		if password != "" {
			if _, err = conn.Do("AUTH", password); err != nil {
				conn.Close()
				return nil, err
			}
		}

		conn.Do("SELECT", conf.DB)
		return conn, nil
	}

	if err := client.pool.Get().Err(); err != nil {
		log.Errorf(fmt.Sprintf("redis connect failed err:%v", err))
		return nil
	}
	return client
}

func GetClient() *RedisClient {
	return client
}

func (r *RedisClient) GetPool() *redis.Pool {
	return r.pool
}

// Do 执行 redis 命令
// NOTE 除非有必要(比如在一个函数内部需要执行多次 redis 操作), 否则请用该函数执行所有的操作, 这样能有效避免忘记释放资源.
func (r *RedisClient) Do(ctx context.Context, commandName string, args ...interface{}) (reply interface{}, err error) {
	// 简单加上链路追踪,后续可封装成traceConn
	conn := r.pool.Get()
	defer conn.Close()

	reply, err = conn.Do(commandName, args...)
	return
}

// 获取指定数量的队列数据
// listName:队列名称
// count:队列元素个数
// 返回:[]string
func (r *RedisClient) GetRangeFromQueue(ctx context.Context, listName string, count int64) []string {
	items, err := redis.Strings(r.Do(ctx, "LRANGE", listName, 0, count))
	if err != nil {
		log.Errorf("获取REDIS LIST 数据失败：" + listName + ";" + err.Error())
		return nil
	}
	return items
}

func (r *RedisClient) ListRange(ctx context.Context, listName string, from int64, to int64) []string {
	items, err := redis.Strings(r.Do(ctx, "LRANGE", listName, from, to))
	if err != nil {
		log.Errorf("获取REDIS LIST 数据失败：" + listName + ";" + err.Error())
		return nil
	}
	return items
}

// ExistsKey 判断redis key是否存在
func (r *RedisClient) ExistsKey(ctx context.Context, key string) bool {
	exists, err := redis.Bool(r.Do(ctx, "EXISTS", key))
	if err != nil {
		log.Errorf("判断键是否存在：" + err.Error())
		return false
	}
	return exists
}

// GetValByKey 依据KEY获取REDIS数据
func (r *RedisClient) GetValByKey(ctx context.Context, key string) string {
	obj, err := redis.String(r.Do(ctx, "GET", key))
	if err != nil {
		if err == redis.ErrNil {
			return ""
		}
		log.Errorf("获取REDIS数据失败：" + key + ";" + err.Error())
		return ""
	}
	return obj
}

// SetRedisItem 设置redis数据key-value
func (r *RedisClient) SetRedisItem(ctx context.Context, key string, val string) {
	_, err := r.Do(ctx, "SET", key, val)
	if err != nil {
		log.Errorf("设置REDIS数据失败：" + key + ";" + val + err.Error())
	}
}

// SetRedisItemWithExpireTime 设置redis数据key-value,并设置过期时间expiredTime，单位s
func (r *RedisClient) SetRedisItemWithExpireTime(ctx context.Context, key string, val string, expiredTime int64) bool {
	_, err := r.Do(ctx, "SET", key, val, "EX", strconv.FormatInt(expiredTime, 10))
	if err != nil {
		log.Errorf("HASH设置REDIS数据失败：" + err.Error())
		return false
	}
	return true
}

// DeletedRedisItem 根据key删除string
func (r *RedisClient) DeletedRedisItem(ctx context.Context, key string) bool {
	_, err := r.Do(ctx, "DEL", key)
	if err != nil {
		log.Errorf("Del REDIS string 失败：" + err.Error())
		return false
	}
	return true
}

func (r *RedisClient) BatchDel(ctx context.Context, keys []string) (count int, err error) {
	// 获取一条Redis连接
	var args []interface{}
	for _, v := range keys {
		args = append(args, v)
	}
	if err != nil {
		return count, err
	}
	delCount, err := redis.Int(r.Do(ctx, "DEL", args...))
	return delCount, err
}

// PushItemToList 向REDIS队列中添加元素
func (r *RedisClient) PushItemToList(ctx context.Context, list string, items ...string) bool {
	args := []interface{}{
		list,
	}
	for i := 0; i < len(items); i++ {
		args = append(args, items[i])
	}

	_, err := r.Do(ctx, "RPUSH", args...) // 添加队列元素
	if err != nil {
		log.Errorf("QUEYE PUSH数据失败：" + list + ";" + err.Error())
		return false
	}
	return true
}

func (r *RedisClient) GetFirstItemFromList(ctx context.Context, list string) string {
	val, err := redis.String(r.Do(ctx, "LINDEX", list, 0)) // 删除队列中指定元素
	if err != nil {
		log.Errorf("GetItemFromRList REDIS数据失败：" + list + ";" + err.Error())
		return ""
	}
	return val
}

// RemoveItemFromList 从REDIS LIST中删除指定元素
func (r *RedisClient) RemoveItemFromList(ctx context.Context, list string, item string) bool {
	_, err := r.Do(ctx, "LREM", list, 0, item) // 删除队列中指定元素
	if err != nil {
		log.Errorf("RemoveItemFromList REDIS数据失败：" + list + ";" + item + err.Error())
		return false
	}
	return true
}

// DelHashItem 删除redis指定HASH项
func (r *RedisClient) DelHashItem(ctx context.Context, hashKey string, key string) bool {
	_, err := r.Do(ctx, "HDEL", hashKey, key)
	if err != nil {
		log.Errorf("删除HASH项REDIS数据失败：" + hashKey + ":" + key + "：" + err.Error())
		return false
	}
	return true
}

// SetHashItem 设置hash item
func (r *RedisClient) SetHashItem(ctx context.Context, hashKey string, key string, value string) bool {
	_, err := r.Do(ctx, "HSET", hashKey, key, value)
	if err != nil {
		log.Errorf("HASH设置REDIS数据失败：" + hashKey + ":" + key + "：" + value + ";err:" + err.Error())
		return false
	}
	return true
}

// ExistsHashItem 判断是否存在Hash Item
func (r *RedisClient) ExistsHashItem(ctx context.Context, hashKey string, key string) bool {
	data, err := redis.Bool(r.Do(ctx, "HEXISTS", hashKey, key))
	if err != nil {
		log.Errorf("判断HASH ITEM是否存在，REDIS获取数据失败：" + hashKey + ":" + key + "err:" + err.Error())
		return false
	}
	return data
}

// GetHashItem 获取HashItem
func (r *RedisClient) GetHashItem(ctx context.Context, hashKey string, key string) string {
	data, err := redis.String(r.Do(ctx, "HGET", hashKey, key))
	if err != nil {
		if err == redis.ErrNil { // 不存在
			return ""
		}
		log.Errorf("HASH REDIS获取数据失败：" + hashKey + ":" + key + "err:" + err.Error())
		return ""
	}
	return data
}

func (r *RedisClient) HGetAll(ctx context.Context, hashKey string) (fieldValue map[string]string, err error) {
	fieldValue = make(map[string]string)
	values, err := redis.StringMap(r.Do(ctx, "hgetall", hashKey))
	if err != nil {
		return
	}

	for k, v := range values {
		if v == "" {
			continue
		}
		fieldValue[k] = v
	}
	return
}

// 批量获取hash数据
func (r *RedisClient) Hmget(ctx context.Context, hashKey string, fields ...string) (fieldValue map[string]string, err error) {
	fieldValue = make(map[string]string)
	if len(fields) == 0 {
		return
	}
	args := []interface{}{hashKey}
	fieldMap := make(map[int]string)
	for k, field := range fields {
		args = append(args, field)
		fieldMap[k] = field
	}
	values, err := redis.Strings(r.Do(ctx, "hmget", args...))
	if err != nil {
		return
	}

	for k, v := range values {
		if v == "" {
			continue
		}
		fieldValue[fieldMap[k]] = v
	}
	return
}

func (r *RedisClient) Hmset(ctx context.Context, hashName string, fieldValue map[interface{}]interface{}) (err error) {
	if len(fieldValue) == 0 {
		return
	}
	args := []interface{}{hashName}

	for field, value := range fieldValue {
		args = append(args, field, value)
	}
	_, err = r.Do(ctx, "hmset", args...)
	return
}

// Lock 分布式上锁
func (r *RedisClient) Lock(ctx context.Context, key, val string, seconds int) bool {
	result, err := redis.String(r.Do(ctx, "SET", key, val, "NX", "EX", seconds))
	if err != nil && err != redis.ErrNil {
		log.Errorf(fmt.Sprintf("set nx ex redis fail, data[:%s:%s], error:%s", key, val, err.Error()))
		return false
	}
	if strings.EqualFold(result, "ok") {
		return true
	}
	return false
}

// UnLock 分布式解锁
func (r *RedisClient) UnLock(ctx context.Context, key string) bool {
	if _, err := r.Do(ctx, "DEL", key); err != nil {
		log.Errorf(fmt.Sprintf("del redis fail, key[%s], error:%s", key, err.Error()))
		return false
	}
	return true
}

// Expire 对key增加过期时间
func (r *RedisClient) Expire(ctx context.Context, key string, seconds int32) bool {
	data, err := redis.Bool(r.Do(ctx, "Expire", key, seconds))
	if err != nil {
		log.Errorf(fmt.Sprintf("expire redis fail, key[%s], error:%s", key, err.Error()))
		return false
	}

	return data
}

// SisMemberSetsItem 查询成员是否存在该set
func (r *RedisClient) SisMemberSetsItem(ctx context.Context, key string, item string) bool {
	exist, err := redis.Bool(r.Do(ctx, "SISMEMBER", key, item))
	if err != nil {
		return false
	}
	return exist
}

func (r *RedisClient) Hincrby(ctx context.Context, hashName string, field interface{}, increment interface{}) (int64, error) {
	return redis.Int64(r.Do(ctx, "Hincrby", hashName, field, increment))
}

func (r *RedisClient) HincrbyFloat(ctx context.Context, hashName string, field interface{}, increment interface{}) (float64, error) {
	return redis.Float64(r.Do(ctx, "hIncrByFloat", hashName, field, increment))
}

func (r *RedisClient) HgetAll(ctx context.Context, hashName string) (value map[string]string, err error) {
	return redis.StringMap(r.Do(ctx, "hgetall", hashName))
}

func (r *RedisClient) Keys(ctx context.Context, prefixKey string) (keys []string, err error) {
	keys, err = redis.Strings(r.Do(ctx, "keys", fmt.Sprintf("%v*", prefixKey)))
	return
}

// Mget 批量查询key
func (r *RedisClient) Mget(ctx context.Context, keys []string) (vals []string, err error) {
	var args []interface{}
	for _, v := range keys {
		args = append(args, v)
	}
	vals, err = redis.Strings(r.Do(ctx, "MGET", args...))
	if err != nil {
		if err == redis.ErrNil {
			return vals, nil
		}
		log.Errorf(fmt.Sprintf("获取REDIS MGet数据失败：%v;%s", keys, err.Error()))
		return
	}
	return
}

// Mset 批量写入
func (r *RedisClient) Mset(ctx context.Context, kvs map[string]string) bool {
	var args []interface{}
	if len(kvs) > 100 {
		return false
	}
	for k, v := range kvs {
		args = append(args, k, v)
	}
	_, err := r.Do(ctx, "MSET", args...)
	if err != nil {
		log.Errorf("REDIS MSET 数据失败：")
		return false
	}
	return true
}

func (r *RedisClient) GetBit(ctx context.Context, key string, offset int) (value int, err error) {
	value, err = redis.Int(r.Do(ctx, "GETBIT", key, offset))
	return value, err
}

func (r *RedisClient) SetBit(ctx context.Context, key string, offset, value int) (err error) {
	// 获取一条Redis连接
	_, err = redis.Int(r.Do(ctx, "SETBIT", key, offset, value))
	return err
}

func (r *RedisClient) MGetBit(ctx context.Context, key string, offsets []int) (value []int, err error) {
	args := []interface{}{key}
	for i := range offsets {
		args = append(args, "GET", "u1", offsets[i])
	}

	value, err = redis.Ints(r.Do(ctx, "BITFIELD", args...))
	if err != nil {
		return nil, err
	}
	return value, err
}

// ZrangeByScore Returns all the elements in the sorted set at key with a score between min and max (including elements with score equal to min or max).
// The elements are considered to be ordered from low to high scores.
func (r *RedisClient) ZrangeByScore(ctx context.Context, key string, min float64, max float64, offset, count int64) (valSlice []string, err error) {
	// zrangebyscore key min max [WITHSCORES] [LIMIT offset count]
	// member1, score1, member2, score2, member3, score3
	valSlice, err = redis.Strings(r.Do(ctx, "ZRANGEBYSCORE", key, min, max, "WITHSCORES", "LIMIT", offset, count))
	return valSlice, err
}

func (r *RedisClient) ZrangeByScoreInt64(ctx context.Context, key string, min int64, max int64, offset, count int64) (valSlice []string, err error) {
	// zrangebyscore key min max [WITHSCORES] [LIMIT offset count]
	// member1, score1, member2, score2, member3, score3
	valSlice, err = redis.Strings(r.Do(ctx, "ZRANGEBYSCORE", key, min, max, "WITHSCORES", "LIMIT", offset, count))
	return valSlice, err
}

func (r *RedisClient) ZRevRangeWithScore(ctx context.Context, key string, start, stop int64) (valSlice []string, err error) {
	// zrangebyscore key min max [WITHSCORES] [LIMIT offset count]
	// member1, score1, member2, score2, member3, score3
	valSlice, err = redis.Strings(r.Do(ctx, "ZrevRange", key, start, stop, "WITHSCORES"))
	return valSlice, err
}

func (r *RedisClient) ZRangeWithScore(ctx context.Context, key string, start, stop int64) (valSlice []string, err error) {
	// zrangebyscore key min max [WITHSCORES] [LIMIT offset count]
	// member1, score1, member2, score2, member3, score3
	valSlice, err = redis.Strings(r.Do(ctx, "ZRange", key, start, stop, "WITHSCORES"))
	return valSlice, err
}

// ZremBatch Removes the specified members from the sorted set stored at key. Non existing members are ignored.
// https://redis.io/commands/zrem
// An error is returned when key exists and does not hold a sorted set.
func (r *RedisClient) ZremBatch(ctx context.Context, key string, member []string) (delCount int, err error) {
	// 获取一条Redis连接
	var args []interface{}
	args = append(args, key)
	for _, v := range member {
		args = append(args, v)
	}
	delCount, err = redis.Int(r.Do(ctx, "ZREM", args...))
	return delCount, err
}

// Deprecated: Use ZremBatch(ctx context.Context, key string, member []string) (delCount int, err error) instead.
func (r *RedisClient) ZremMulti(ctx context.Context, key string, member []string) (int64, error) {
	delCnt, err := r.ZremBatch(ctx, key, member)
	return int64(delCnt), err
}

// Zcard returns the sorted set cardinality (number of elements) of the sorted set stored at key.
// Integer reply: the cardinality (number of elements) of the sorted set, or 0 if key does not exist.
func (r *RedisClient) Zcard(ctx context.Context, key string) (count int64, err error) {
	count, err = redis.Int64(r.Do(ctx, "ZCARD", key))
	return count, err
}

// Zclear delete all members in a zset
func (r *RedisClient) Zclear(ctx context.Context, key string) (err error) {
	_, err = r.Do(ctx, "ZCLEAR", key)
	return err
}

// Zcount returns the number of elements in the sorted set at key with a score between min and max.
func (r *RedisClient) Zcount(ctx context.Context, key string, min float64, max float64) (count int64, err error) {
	// ZCOUNT key min max
	count, err = redis.Int64(r.Do(ctx, "ZCOUNT", key, min, max))
	return count, err
}

// ZcountInt64 Zcount returns the number of elements in the sorted set at key with a score between min and max.
func (r *RedisClient) ZcountInt64(ctx context.Context, key string, min int64, max int64) (count int64, err error) {
	// ZCOUNT key min max
	count, err = redis.Int64(r.Do(ctx, "ZCOUNT", key, min, max))
	return count, err
}

// ZremRangeByScore removes all elements in the sorted set stored at key with a score between min and max (inclusive).
// Integer reply: the number of elements removed.
func (r *RedisClient) ZremRangeByScore(ctx context.Context, key string, min float64, max float64) (count int64, err error) {
	//  ZREMRANGEBYSCORE key min max
	count, err = redis.Int64(r.Do(ctx, "ZREMRANGEBYSCORE", key, min, max))
	return count, err
}

func (r *RedisClient) ZremRangeByScoreInt64(ctx context.Context, key string, min int64, max int64) (count int64, err error) {
	//  ZREMRANGEBYSCORE key min max
	count, err = redis.Int64(r.Do(ctx, "ZREMRANGEBYSCORE", key, min, max))
	return count, err
}

// ZaddBatch ZADD key score member [score member ...]
func (r *RedisClient) ZaddBatch(ctx context.Context, key string, scoreMemberPairs []string) (value int64, err error) {
	// 获取一条Redis连接
	var args []interface{}
	args = append(args, key)
	keyScoreLne := len(scoreMemberPairs)
	for i := 0; i < keyScoreLne; i += 2 { // key-score pari
		score := scoreMemberPairs[i]
		member := scoreMemberPairs[i+1]
		args = append(args, score)
		args = append(args, member)
	}
	value, err = redis.Int64(r.Do(ctx, "ZADD", args...))
	return value, err
}

// Zrange 返回 指定区间内，带有 score 值(可选)的有序集成员的列表
func (r *RedisClient) Zrange(ctx context.Context, key string, start, stop int64) (valSlice []string, err error) {
	// 获取一条Redis连接
	valSlice, err = redis.Strings(r.Do(ctx, "ZRANGE", key, start, stop, "WITHSCORES"))
	return valSlice, err
}

// Zadd 返回 被成功添加的新成员的数量，不包括那些被更新的、已经存在的成员
func (r *RedisClient) Zadd(ctx context.Context, key string, score int64, member string) (value int64, err error) {
	// 获取一条Redis连接
	value, err = redis.Int64(r.Do(ctx, "ZADD", key, score, member))
	return value, err
}

// Zrem 返回 被成功移除的成员的数量，不包括被忽略的成员。
func (r *RedisClient) Zrem(ctx context.Context, key, member string) (value int64, err error) {
	// 获取一条Redis连接
	value, err = redis.Int64(r.Do(ctx, "ZREM", key, member))
	return value, err
}

// Publish 发布消息
func (r *RedisClient) Publish(ctx context.Context, topic string, message []byte) (bSuccess bool) {
	var err error

	// 执行推送
	bSuccess, err = redis.Bool(r.Do(ctx, "publish", topic, message))
	if err != nil {
		return false
	}

	return bSuccess
}

// HandlerFunc 处理函数
type HandlerFunc func(data []byte)

func (r *RedisClient) Consumer(ctx context.Context, topics []interface{}, subChannelMap map[string]HandlerFunc) (err error) {
	defer utils.ExceptionCatch()
lable:
	conn := r.GetPool().Get()
	defer conn.Close()
	psc := redis.PubSubConn{
		Conn: conn,
	}

	err = psc.Subscribe(topics...)
	if err != nil {
		fmt.Println("failed to subscribe message,err:", err)
		time.Sleep(time.Second * 10)
		goto lable
	}
	for {
		switch msg := psc.Receive().(type) {
		case redis.Message:
			if subChannelMap == nil {
				continue
			}

			if h, bExist := subChannelMap[msg.Channel]; bExist && (h != nil) {
				h(msg.Data)
			} else {
				fmt.Println("SubscriberPublish receive channel", msg.Channel)
			}
		case error:
			fmt.Println("subscriber receive err:", err, "restart it")
			time.Sleep(time.Second * 10)
			goto lable
		}
	}
}
